package com.rocketmq.example;

import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;

public class E05_BroadcastingMessage01 {

    private static final String TOPIC = "BroadcastingMessageTopic";

    private static final String GROUP = "Broadcasting";

    /**
     * 广播模式
     * 广播模式是多个消费者可以同时接到同一条消息
     * consumer.setMessageModel(MessageModel.BROADCASTING);
     */

    public static void main(String[] args) {
        consumer1();
    }

    public static void consumer1() {

        try {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(GROUP);
            // 指定NameServer的地址
            consumer.setNamesrvAddr(ExampleMQConfig.ENDPOINT.getName());

            // 订阅这个topic下的所有的消息
            consumer.subscribe(TOPIC, "*");

            // 设置为广播模式
            consumer.setMessageModel(MessageModel.BROADCASTING);

            // 注册一个消费的监听器，当有消息的时候，会回调这个监听器来消费消息
            consumer.registerMessageListener(new MessageListenerConcurrently() {

                @Override
                // setConsumeMessageBatchMaxSize 就是指 List<MessageExt> msgs 的大小
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                        ConsumeConcurrentlyContext context) {
                    for (MessageExt msg : msgs) {
                        System.out.printf("消费消息:%s", GROUP + new String(msg.getBody()) + "\n");
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
            System.err.println("消费者启动成功");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
